package com.gjy.kafka.resolve.producer;

import com.gjy.kafka.resolve.config.KafkaProducerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * 使用 enable.idempotence 实现“消息不丢失 + 不重复写入”
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 15:19:37
 */
public class IdempotentProducer implements AutoCloseable {

    private static final Logger logger = LoggerFactory.getLogger(IdempotentProducer.class);
    private final Producer<String, String> producer;

    public IdempotentProducer() {
        Properties props = KafkaProducerConfig.buildConfig(null); // null → 普通幂等生产者
        this.producer = new KafkaProducer<>(props);
    }

    /**
     * 同步发送，适用于关键业务（必须确认成功后才返回）
     */
    public void sendSync(String topic, String key, String value) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        try {
            RecordMetadata meta = producer.send(record).get(); // 阻塞等待 ack
            logger.info("发送成功: topic={}, partition={}, offset={}", meta.topic(),
                    meta.partition(), meta.offset());
        } catch (Exception e) {
            logger.error("发送失败，业务可自行重试或告警", e);
            // 这里的重试可以依赖 producer 内部的 retries 或外层自定义
        }
    }

    /**
     * 异步发送（业务自行处理回调）
     */
    public void sendAsync(String topic, String key, String value) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                logger.info("异步发送成功: topic={}, partition={}, offset={}",
                        metadata.topic(), metadata.partition(), metadata.offset());
            } else {
                logger.error("异步发送失败", exception);
                // 可根据异常类型自行决定是否重试
            }
        });
    }

    @Override
    public void close() {
        producer.flush();
        producer.close();
    }

}
